package com.atguigu.bigdata.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

object SparkStreamingStudy_updateStateByKey1 {

  def main(args: Array[String]): Unit = {
    // 定义更新状态方法，参数values为当前数字，state为以往数字之和
    val updateFunc = (values: Seq[Int], state: Option[Int]) => {
      val currentCount = values.foldLeft(0)(_+_)
      val previousCount = state.getOrElse(0)
      Some(currentCount + previousCount)
    }

    val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
    val ssc = new StreamingContext(conf, Seconds(3))
    //ssc.checkpoint("hdfs://hadoop102:9000/streamCheck")
    ssc.checkpoint("E://checkpoint")
    // Create a DStream that will connect to hostname:port, like hadoop102:9999
    val lines = ssc.socketTextStream("10.21.13.181", 9999)

    // Split each line
    val words = lines.flatMap(_.split(" "))

    // updateStateByKey是根据key来计算，给所有的数字同一个key就可以实现不断的累加数字的功能
    val pairs = words.map(word => ("SUM", word.trim.toInt))

    // 使用updateStateByKey来更新状态，统计从运行开始以来单词总的次数
    val stateDstream = pairs.updateStateByKey[Int](updateFunc)
    stateDstream.print()
    stateDstream.foreachRDD(rdd=>{
      //rdd[(string,int)]转换为rdd[Int]
      //rddSum[Int]
      val rddSum = rdd.map{(x)=>(x._2)}
      println(rddSum.first())
    })
    //val wordCounts = pairs.reduceByKey(_ + _)

    // Print the first ten elements of each RDD generated in this DStream to the console
    //wordCounts.print()
    ssc.start()             // Start the computation
    ssc.awaitTermination()  // Wait for the computation to terminate
    //ssc.stop()
  }
}
